package com.nl.networkflow;


import com.nl.bean.input.UserBehavior;
import java.net.URL;
import java.sql.Timestamp;
import java.util.HashSet;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author shihb
 * @date 2019/12/20 11:30
 * 独立访客数(uv)的统计
 */
public class UniqueVisitor {

  public static void main(String[] args) throws Exception {
    //1.创建执行环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);


    //2.source
    URL resource = PageView.class.getClassLoader().getResource("UserBehavior.csv");
    DataStreamSource<String> source = env
        .readTextFile(resource.getPath());
    //3.transform算子
    SingleOutputStreamOperator<Tuple2<Timestamp, Long>> dataStream = source
        .map(s->{
          String[] arr = s.split(",");
          long userId = Long.parseLong(arr[0].trim());
          long itemId = Long.parseLong(arr[1].trim());
          int categoryId = Integer.parseInt(arr[2].trim());
          String behavior = arr[3].trim();
          long timestamp = Long.parseLong(arr[4].trim());
          return UserBehavior.of(userId, itemId, categoryId, behavior, timestamp);
        })
        .assignTimestampsAndWatermarks(
            new BoundedOutOfOrdernessTimestampExtractor<UserBehavior>(Time.seconds(0)) {
              @Override
              public long extractTimestamp(UserBehavior element) {
                return element.getTimestamp() * 1000;
              }
            }
        )
        .filter(userBehavior->"pv".equals(userBehavior.getBehavior()))
        // 非key的窗口函数
        .timeWindowAll(Time.hours(1))
        // 窗口处理函数,去重处理
        .apply(new AllWindowFunction<UserBehavior, Tuple2<Timestamp, Long>, TimeWindow>() {
          @Override
          public void apply(TimeWindow window, Iterable<UserBehavior> values,
              Collector<Tuple2<Timestamp, Long>> out) throws Exception {
            // 定义一个 set用于用户去重
            HashSet<Long> set = new HashSet<Long>();
            for(UserBehavior ub:values){
              set.add(ub.getUserId());
            }
            out.collect(new Tuple2<Timestamp,Long>(new Timestamp(window.getEnd()),
                (long) set.size()));

          }
        });

    dataStream.print("uv count:");
    env.execute("uv job");

  }

}
